package org.budo.dubbo.protocol.async;

import java.util.Date;

import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.budo.dubbo.protocol.async.util.AsyncLogUtil;
import org.budo.dubbo.protocol.async.util.DestinationUtil;
import org.budo.dubbo.protocol.async.util.RepositoryUtil;
import org.budo.support.lang.util.ThreadUtil;
import org.budo.support.spring.util.SpringUtil;

import com.alibaba.dubbo.rpc.Exporter;

import lombok.NoArgsConstructor;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
@NoArgsConstructor
@ToString
public class PopThread extends Thread {
    /**
     * 线程启动时间
     */
    private volatile Date _startAt = null;

    /**
     * 线程结束时间
     */
    private volatile Date _stopAt = null;

    /**
     * 线程是否循环跑的开关
     */
    private boolean _threadRun = true;

    private Exporter<?> _exporter;

    private AbstractAsyncRepository _asyncRepository;

    public PopThread(Exporter<?> exporter) {
        log.info("#167 PopThread init, this=" + this + ", destinations=" + DestinationUtil.destinations());

        // 启动时间
        this._startAt = new Date();

        this._exporter = exporter;

        String threadName = DestinationUtil.getDestinationName(exporter.getInvoker()) + "-PopThread";

        this.setName(threadName);
        this.setDaemon(true);
    }

    @Override
    public void run() {
        while (PopThread.this._threadRun) { // 希望跑
            if (null == this.asyncRepository()) { // 还没准备好?
                ThreadUtil.sleep(2000); // 等两秒继续
                continue;
            }

            ThreadUtil.sleep(5000); // 准备好了也延迟5秒再来
            long start = System.currentTimeMillis();

            // 执行
            PopThread.this.popMessage_1();

            if (log.isDebugEnabled()) {
                log.debug("#112 _popMessage, 耗时 " + (System.currentTimeMillis() - start) + ", this=" + this);
            }
        }

        // 跳出循环时才会执行到这里
        log.warn("#164 outing thread now, this=" + this);
        PopThread.this._stopAt = new Date();
    }

    public PopThread startThread() {
        log.info("#59 startThread, this=" + this);
        this.start(); // 启动线程

        return this;
    }

    public void stopThread() {
        Throwable e = new Throwable("AbstractAsyncRepository.destroy");
        log.warn("#107 destroy, this=" + this, e);

        // 不继续循环拉数据
        this._threadRun = false;

        // 等待跑完当前一个
        int i = 0;
        while (null == this._stopAt) {
            i++;
            ThreadUtil.sleep(500); // 缓半秒钟

            if (i < 20 || i % 2 == 0) { // 每2秒一次日志
                log.warn("#130 thread stoping, popThread=" + this);
            }
        }

        log.warn("#126 thread stoped, popThread=" + this);
    }

    private AbstractAsyncRepository asyncRepository() {
        if (null != this._asyncRepository) {
            return this._asyncRepository;
        }

        String beanName = RepositoryUtil.getAsyncRepositoryBeanName(this._exporter.getInvoker());
        AbstractAsyncRepository asyncRepository = (AbstractAsyncRepository) SpringUtil.getBeanCached(beanName);

        if (null != asyncRepository) {
            return this._asyncRepository = asyncRepository;
        }

        // 还没准备好?
        log.error("#70 abstractAsyncRepository=" + asyncRepository + ", asyncRepositoryBeanName=" + beanName + ", this=" + this);
        return null;
    }

    private void popMessage_1() {
        // 如果没有消息，就休息一段时间
        Integer popIdleSeconds = this.asyncRepository().getPopIdleSeconds();

        try {
            Integer messageCount = this.asyncRepository().popMessage(this._exporter);
            if (messageCount > 0) {
                if (log.isDebugEnabled()) {
                    log.debug("#130 _popMessage, messageCount=" + messageCount);
                }

                return; // 有消息 回去继续
            }

            if (log.isDebugEnabled()) {
                log.debug("#140 sleep, messageCount=" + messageCount + ", popIdleSeconds=" + popIdleSeconds + ", this=" + this);
            }

            // 休息
            ThreadUtil.sleep(1000 * popIdleSeconds);
        } catch (Throwable e) {
            AsyncLogUtil.error("#216 doPopMessage error, sleep 5 * popIdleSeconds ", e);

            ThreadUtil.sleep(5000 * popIdleSeconds);
        }
    }
}